package com.qupeng.demo.kafka.kafkaspringbootstreams.timewindow;

import com.qupeng.demo.kafka.kafkaspringbootstreams.common.kafka.streams.JsonSerde;
import com.qupeng.demo.kafka.kafkaspringbootstreams.common.kafka.utils.KafkaAdminTools;
import com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.model.AgentCase;
import com.qupeng.demo.kafka.kafkaspringbootstreams.demo.CaseState;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
public class TimeWindowStreams {

    @Autowired
    private KafkaAdminTools kafkaAdminTools;

    @Bean
    public KStream<Windowed<String>, Long> timeWindowTestingKStream(StreamsBuilder myKStreamBuilder) throws Exception {
        CreateTopicsResult resultInput = kafkaAdminTools.createTopic("time-window-testing-input", 3, (short)3);
        CreateTopicsResult resultOutput = kafkaAdminTools.createTopic("time-window-testing-output", 3, (short)3);
        KStream<String, AgentCase> agentCases = myKStreamBuilder.stream("time-window-testing-input", Consumed.with(Serdes.String(), new JsonSerde(AgentCase.class)));
        KStream<Windowed<String>, Long> caseStatistics = agentCases
                .groupByKey(Grouped.with(Serdes.String(), new JsonSerde<>(CaseState.class)))
                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(4)).advanceBy(Duration.ofSeconds(60)))
                .count()
                .toStream();
        caseStatistics.print(Printed.toSysOut());
        caseStatistics.process(WindowedLongProcessor::new);
        caseStatistics.to("time-window-testing-output", Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class, 4*60L*1000)));
        return caseStatistics;
    }
}
